package com.heiye.quiz.biz.service.impl;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.util.RandomUtil;
import com.github.benmanes.caffeine.cache.Cache;
import com.github.benmanes.caffeine.cache.Caffeine;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.heiye.count.api.dto.FindQuizCountsByIdRspDTO;
import com.heiye.count.api.dto.FindUserCountsByIdReqDTO;
import com.heiye.count.api.dto.FindUserCountsByIdRspDTO;
import com.heiye.framework.common.exception.BizException;
import com.heiye.framework.common.response.Response;
import com.heiye.framework.common.util.JsonUtils;
import com.heiye.quiz.api.dto.req.CheckNoteIsExistReqDTO;
import com.heiye.quiz.api.dto.req.FindQuizCreatorReqDTO;
import com.heiye.quiz.api.dto.resp.CheckQuizIsExistRespDTO;
import com.heiye.quiz.api.dto.resp.FindQuizCreatorRespDTO;
import com.heiye.quiz.biz.constant.MQConstants;
import com.heiye.quiz.biz.constant.RedisKeyConstants;
import com.heiye.quiz.biz.domain.dataobject.QuizDO;
import com.heiye.quiz.biz.domain.dataobject.QuizLikeDO;
import com.heiye.quiz.biz.domain.mapper.QuizDOMapper;
import com.heiye.quiz.biz.domain.mapper.QuizLikeDOMapper;
import com.heiye.quiz.biz.enums.*;
import com.heiye.quiz.biz.model.dto.LikeUnlikeQuizMqDTO;
import com.heiye.quiz.biz.model.dto.QuizOperateMqDTO;
import com.heiye.quiz.biz.model.vo.req.*;
import com.heiye.quiz.biz.model.vo.rsep.FindQuizDetailRspVO;
import com.heiye.quiz.biz.model.vo.rsep.FindQuizIsLikeRspVO;
import com.heiye.quiz.biz.model.vo.rsep.FindQuizzesRspVO;
import com.heiye.quiz.biz.model.vo.rsep.QuizItemRspVO;
import com.heiye.quiz.biz.rpc.*;
import com.heiye.quiz.biz.service.QuizService;
import com.heiye.relation.api.dto.FindQuizFansTop3ByIdsRspDTO;
import com.heiye.user.api.dto.resp.FindUserByIdRspDTO;
import jakarta.annotation.Resource;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.heiye.framework.biz.context.thread.LoginUserContextHolder;
import org.springframework.core.io.ClassPathResource;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.scripting.support.ResourceScriptSource;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.time.LocalDateTime;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;
import java.util.stream.Collectors;

/**
 * @author: heiye
 * @date: 2025/03/17 下午10:20
 * @version: v1.0.0
 * @description: 问题业务
 */
@Slf4j
@Service
public class QuizServiceImpl implements QuizService {

    @Resource
    private DistributedIdGeneratorRpcService distributedIdGeneratorRpcService;

    @Resource
    private KeyValueRpcService keyValueRpcService;

    @Resource
    private QuizDOMapper quizDOMapper;

    @Resource
    private UserRpcService userRpcService;

    @Resource(name = "taskExecutor")
    private ThreadPoolTaskExecutor threadPoolTaskExecutor;

    @Resource
    private RedisTemplate<String, String> redisTemplate;

    @Resource
    private RocketMQTemplate rocketMQTemplate;

    @Resource
    private QuizLikeDOMapper quizLikeDOMapper;

    @Resource
    private CountRpcService countRpcService;

    @Resource
    private RelationRpcService relationRpcService;

    private static final Cache<Long, String> LOCAL_CACHE = Caffeine.newBuilder()
            // 设置初始容量为 10000 个条目
            .initialCapacity(10000)
            // 设置缓存的最大容量为 10000 个条目
            .maximumSize(10000)
            // 设置缓存条目在写入后 1 小时过期
            .expireAfterWrite(1, TimeUnit.HOURS)
            .build();

    /**
     * 问题发布
     *
     * @param publishQuizReqVO
     * @return
     */
    @Override
    public Response<?> publishQuiz(PublishQuizReqVO publishQuizReqVO) {

        // 提问内容是否为空，默认值为 true，即空
        boolean isContentEmpty = true;

        // RPC: 调用分布式 ID 生成服务，生成笔记 ID
        String snowflakeId = distributedIdGeneratorRpcService.getQuizId();

        // 提问内容 UUID
        String contentUuid = null;

        // 提问内容
        String content = publishQuizReqVO.getContent();

        // 若用户填写了提问内容
        if (StringUtils.isNotBlank(content)) {
            // 内容是否为空，置为 false，即不为空
            isContentEmpty = false;
            // 生成提问内容 UUID
            contentUuid = UUID.randomUUID().toString();
            // RPC: 调用 KV 键值服务，存储短文本
            boolean isSavedSuccess = keyValueRpcService.saveQuizContent(contentUuid, content);
            // 若存储失败，抛出业务异常，提示用户发布问题失败
            if (!isSavedSuccess) {
                return Response.fail(ResponseCodeEnum.QUIZ_PUBLISH_FAIL);
            }
        }

        // 发布者用户 ID
        Long creatorId = LoginUserContextHolder.getUserId();

        // 构建提问 DO 对象
        QuizDO quizDO = QuizDO.builder()
                .id(Long.valueOf(snowflakeId))
                .title(publishQuizReqVO.getTitle())
                .isContentEmpty(isContentEmpty)
                .creatorId(creatorId)
                .isAnswer(Boolean.FALSE)
                .usefulAnswerId(null)
                .contentUuid(contentUuid)
                .languageId(publishQuizReqVO.getLanguageId())
                .createTime(LocalDateTime.now())
                .updateTime(LocalDateTime.now())
                .status(QuizStatusEnum.NORMAL.getCode())
                .build();

        try {
            // 笔记入库存储
            quizDOMapper.insert(quizDO);
        } catch (Exception e) {
            log.error("==> 笔记存储失败", e);

            // RPC: 笔记保存失败，则删除笔记内容
            keyValueRpcService.deleteQuizContent(contentUuid);
        }

        // 发送 MQ
        // 构建消息体 DTO
        QuizOperateMqDTO quizOperateMqDTO = QuizOperateMqDTO.builder()
                .creatorId(creatorId)
                .quizId(Long.valueOf(snowflakeId))
                // 发布笔记
                .type(QuizOperateEnum.PUBLISH.getCode())
                .build();

        // 构建消息对象，并将 DTO 转成 Json 字符串设置到消息体中
        Message<String> message = MessageBuilder.withPayload(JsonUtils.toJsonString(quizOperateMqDTO))
                .build();

        // 通过冒号连接, 可让 MQ 发送给主题 Topic 时，携带上标签 Tag
        String destination = MQConstants.TOPIC_QUIZ_OPERATE + ":" + MQConstants.TAG_QUIZ_PUBLISH;

        // 异步发送 MQ 消息，提升接口响应速度
        rocketMQTemplate.asyncSend(destination, message, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("==> 【问题发布】MQ 发送成功，SendResult: {}", sendResult);
            }

            @Override
            public void onException(Throwable throwable) {
                log.error("==> 【问题发布】MQ 发送异常: ", throwable);
            }
        });

        return Response.success();
    }

    /**
     * 问题详情
     *
     * @param findQuizDetailReqVO
     * @return
     */
    @Override
    @SneakyThrows
    public Response<FindQuizDetailRspVO> findQuizDetail(FindQuizDetailReqVO findQuizDetailReqVO) {
        // 查询的问题 ID
        Long quizId = findQuizDetailReqVO.getId();

        // 先从本地缓存中查询
        String findQuizDetailRspVOStrLocalCache = LOCAL_CACHE.getIfPresent(quizId);
        if (StringUtils.isNotBlank(findQuizDetailRspVOStrLocalCache)) {
            FindQuizDetailRspVO findQuizDetailRspVO = JsonUtils.parseObject(findQuizDetailRspVOStrLocalCache, FindQuizDetailRspVO.class);
            log.info("==> 命中了本地缓存；{}", findQuizDetailRspVOStrLocalCache);
            return Response.success(findQuizDetailRspVO);
        }

        // 从 Redis 缓存中获取
        String buildQuizDetailRedisKey = RedisKeyConstants.buildQuizDetailKey(quizId);
        String quizDetailRedisValue = redisTemplate.opsForValue().get(buildQuizDetailRedisKey);

        // 若缓存中有该笔记的数据，则直接返回
        if (StringUtils.isNotBlank(quizDetailRedisValue)) {
            FindQuizDetailRspVO findQuizDetailRspVO = JsonUtils.parseObject(quizDetailRedisValue, FindQuizDetailRspVO.class);
            // 异步线程中将用户信息存入本地缓存
            threadPoolTaskExecutor.submit(() -> {
                // 写入本地缓存
                LOCAL_CACHE.put(
                        quizId,
                        Objects.isNull(findQuizDetailRspVO) ? "null" : JsonUtils.toJsonString(findQuizDetailRspVO)
                );
            });
            return Response.success(findQuizDetailRspVO);
        }

        // 查询问题
        QuizDO quizDO = quizDOMapper.selectByPrimaryKey(quizId);

        // 如果问题不存在，抛出业务异常，提示用户问题不存在
        if (Objects.isNull(quizDO)) {
            threadPoolTaskExecutor.execute(() -> {
                // 防止缓存穿透，将空数据存入 Redis 缓存 (过期时间不宜设置过长)
                // 保底1分钟 + 随机秒数
                int expireSeconds = 60 + RandomUtil.randomInt(60);
                redisTemplate.opsForValue().set(buildQuizDetailRedisKey, "null", expireSeconds, TimeUnit.SECONDS);
            });
            return Response.fail(ResponseCodeEnum.QUIZ_NOT_FOUND);
        }

        // 并发查询优化
        // RPC: 调用用户服务
        Long creatorId = quizDO.getCreatorId();
        CompletableFuture<FindUserByIdRspDTO> userResultFuture = CompletableFuture
                .supplyAsync(() -> userRpcService.findById(creatorId), threadPoolTaskExecutor);

        // RPC: 调用 K-V 存储服务获取内容
        CompletableFuture<String> contentResultFuture = CompletableFuture.completedFuture(null);
        if (Objects.equals(quizDO.getIsContentEmpty(), Boolean.FALSE)) {
            contentResultFuture = CompletableFuture
                    .supplyAsync(() -> keyValueRpcService.findQuizContent(quizDO.getContentUuid()), threadPoolTaskExecutor);
        }

        CompletableFuture<String> finalContentResultFuture = contentResultFuture;
        CompletableFuture<FindQuizDetailRspVO> resultFuture = CompletableFuture
                .allOf(userResultFuture, finalContentResultFuture)
                .thenApply(s -> {
                    // 获取 Future 返回的结果
                    FindUserByIdRspDTO findUserByIdRspDTO = userResultFuture.join();
                    String content = finalContentResultFuture.join();

                    // 构建返参 VO 实体类
                    return FindQuizDetailRspVO.builder()
                            .id(quizDO.getId())
                            .title(quizDO.getTitle())
                            .creatorId(findUserByIdRspDTO.getId())
                            .creatorName(findUserByIdRspDTO.getDisplayName())
                            .avatar(findUserByIdRspDTO.getPhotoUrl())
                            .nativeLanguage(findUserByIdRspDTO.getNativeLanguage())
                            .interestLanguage(findUserByIdRspDTO.getInterestLanguage())
                            .content(content)
                            .isAnswer(quizDO.getIsAnswer())
                            .usefulAnswerId(quizDO.getUsefulAnswerId())
                            .updateTime(quizDO.getUpdateTime())
                            .build();
                });

        // 获取拼装后的 FindNoteDetailRspVO
        FindQuizDetailRspVO findQuizDetailRspVO = resultFuture.get();

        // 异步将用户信息存入 Redis 缓存，提升响应速度
        threadPoolTaskExecutor.submit(() -> {
            // 过期时间（保底1天 + 随机秒数，将缓存过期时间打散，防止同一时间大量缓存失效，导致数据库压力太大）
            long expireSeconds = 60 * 60 * 24 + RandomUtil.randomInt(60 * 60 * 24);

            redisTemplate.opsForValue().set(buildQuizDetailRedisKey, JsonUtils.toJsonString(findQuizDetailRspVO), expireSeconds, TimeUnit.SECONDS);
        });

        return Response.success(findQuizDetailRspVO);
    }

    /**
     * 问题更新
     *
     * @param updateQuizReqVO
     * @return
     */
    @Override
    @Transactional(rollbackFor = Exception.class)
    public Response<?> updateQuiz(UpdateQuizReqVO updateQuizReqVO) {
        // 问题 ID
        Long quizId = updateQuizReqVO.getId();

        // 查询出问题
        QuizDO selectNoteDO = quizDOMapper.selectByPrimaryKey(quizId);

        // 笔记不存在
        if (Objects.isNull(selectNoteDO)) {
            throw new BizException(ResponseCodeEnum.QUIZ_NOT_FOUND);
        }

        // 当前登录用户 ID
        Long currUserId = LoginUserContextHolder.getUserId();

        // 判断权限：非笔记发布者不允许更新笔记
        if (!Objects.equals(currUserId, selectNoteDO.getCreatorId())) {
            throw new BizException(ResponseCodeEnum.QUIZ_CANT_OPERATE);
        }

        // 删除 Redis 缓存
        String quizDetailRedisKey = RedisKeyConstants.buildQuizDetailKey(quizId);
        String quizBriefRedisKey = RedisKeyConstants.buildQuizBriefKey(quizId);
        redisTemplate.delete(List.of(quizDetailRedisKey, quizBriefRedisKey));

        // 更新问题元数据表 t_quiz
        String content = updateQuizReqVO.getContent();
        QuizDO quizDO = QuizDO.builder()
                .id(quizId)
                .title(updateQuizReqVO.getTitle())
                .isContentEmpty(StringUtils.isBlank(content))
                .updateTime(LocalDateTime.now())
                .build();

        quizDOMapper.updateByPrimaryKey(quizDO);

        // 一致性保证：延迟双删策略
        // 异步发送延时消息
        sendDelayDeleteRedisQuizCacheMQ(quizId);

        // 删除本地缓存
        // LOCAL_CACHE.invalidate(quizId);

        // 同步发送广播模式 MQ，将所有实例中的本地缓存都删除掉
        rocketMQTemplate.syncSend(MQConstants.TOPIC_DELETE_QUIZ_LOCAL_CACHE, quizId);
        log.info("====> MQ：删除问题本地缓存发送成功...");

        // 笔记内容更新
        // 笔记内容是否更新成功
        boolean isUpdateContentSuccess = false;

        // 获取该篇问题内容对应的 UUID
        String contentUuid = selectNoteDO.getContentUuid();

        if (StringUtils.isBlank(content)) {
            // 若笔记内容为空，则删除 K-V 存储
            isUpdateContentSuccess = keyValueRpcService.deleteQuizContent(contentUuid);
        } else {
            // 若将无内容的笔记，更新为了有内容的笔记，需要重新生成 UUID
            contentUuid = StringUtils.isBlank(contentUuid) ? UUID.randomUUID().toString() : contentUuid;
            // 调用 K-V 更新短文本
            isUpdateContentSuccess = keyValueRpcService.saveQuizContent(contentUuid, content);
        }

        // 如果更新失败，抛出业务异常，回滚事务
        if (!isUpdateContentSuccess) {
            throw new BizException(ResponseCodeEnum.QUIZ_UPDATE_FAIL);
        }

        return Response.success();
    }

    /**
     * 删除本地问题缓存
     *
     * @param quizId
     */
    @Override
    public void deleteQuizLocalCache(Long quizId) {
        LOCAL_CACHE.invalidate(quizId);
    }

    /**
     * 问题删除
     *
     * @param deleteQuizReqVO
     * @return
     */
    @Override
    public Response<?> deleteQuiz(DeleteQuizReqVO deleteQuizReqVO) {
        // 问题 ID
        Long quizId = deleteQuizReqVO.getId();

        // 查询出问题
        QuizDO selectQuizDO = quizDOMapper.selectByPrimaryKey(quizId);

        // 笔记不存在
        if (Objects.isNull(selectQuizDO)) {
            throw new BizException(ResponseCodeEnum.QUIZ_NOT_FOUND);
        }

        // 当前登录用户 ID
        Long currUserId = LoginUserContextHolder.getUserId();

        // 判断权限：非笔记发布者不允许更新笔记
        if (!Objects.equals(currUserId, selectQuizDO.getCreatorId())) {
            throw new BizException(ResponseCodeEnum.QUIZ_CANT_OPERATE);
        }

        // 删除缓存
        String quizDetailRedisKey = RedisKeyConstants.buildQuizDetailKey(quizId);
        String quizBriefRedisKey = RedisKeyConstants.buildQuizBriefKey(quizId);
        redisTemplate.delete(Arrays.asList(quizDetailRedisKey, quizBriefRedisKey));

        // 逻辑删除
        QuizDO quizDO = QuizDO.builder()
                .id(quizId)
                .updateTime(LocalDateTime.now())
                .status(QuizStatusEnum.DELETED.getCode())
                .build();

        quizDOMapper.updateByPrimaryKeySelective(quizDO);

        // 延迟双删
        sendDelayDeleteRedisQuizCacheMQ(quizId);

        // 同步发送广播模式 MQ，将所有实例中的本地缓存都删除掉
        rocketMQTemplate.syncSend(MQConstants.TOPIC_DELETE_QUIZ_LOCAL_CACHE, quizId);
        log.info("====> MQ：删除问题本地缓存发送成功...");

        // 发送 MQ
        // 构建消息体 DTO
        QuizOperateMqDTO quizOperateMqDTO = QuizOperateMqDTO.builder()
                .creatorId(selectQuizDO.getCreatorId())
                .quizId(quizId)
                // 删除笔记
                .type(QuizOperateEnum.DELETE.getCode())
                .build();

        // 构建消息对象，并将 DTO 转成 Json 字符串设置到消息体中
        Message<String> message = MessageBuilder.withPayload(JsonUtils.toJsonString(quizOperateMqDTO))
                .build();

        // 通过冒号连接, 可让 MQ 发送给主题 Topic 时，携带上标签 Tag
        String destination = MQConstants.TOPIC_QUIZ_OPERATE + ":" + MQConstants.TAG_QUIZ_DELETE;

        // 异步发送 MQ 消息，提升接口响应速度
        rocketMQTemplate.asyncSend(destination, message, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("==> 【问题删除】MQ 发送成功，SendResult: {}", sendResult);
            }

            @Override
            public void onException(Throwable throwable) {
                log.error("==> 【问题删除】MQ 发送异常: ", throwable);
            }
        });

        return Response.success();
    }

    /**
     * 发送延时删除 Redis 问题缓存 MQ
     *
     * @param noteId
     */
    private void sendDelayDeleteRedisQuizCacheMQ(Long noteId) {
        Message<String> message = MessageBuilder.withPayload(JsonUtils.toJsonString(noteId)).build();

        rocketMQTemplate.asyncSend(MQConstants.TOPIC_DELAY_DELETE_QUIZ_REDIS_CACHE, message, new SendCallback() {
                    @Override
                    public void onSuccess(SendResult sendResult) {
                        log.info("## 延时删除 Redis 问题缓存消息发送成功...");
                    }

                    @Override
                    public void onException(Throwable e) {
                        log.error("## 延时删除 Redis 问题缓存消息发送失败...", e);
                    }
                },
                // 超时时间(毫秒)
                3000,
                // 延迟级别，1 表示延时 1s
                1
        );
    }

    /**
     * 批量查询问题
     *
     * @param findQuizzesReqVO
     * @return
     */
    @Override
    @SneakyThrows
    public Response<FindQuizzesRspVO> findQuizzes(FindQuizzesReqVO findQuizzesReqVO) {
        // 笔记ID
        Long cursor = findQuizzesReqVO.getCursor();
        // 需要推送的目标语言 ID
        Long languageId = findQuizzesReqVO.getLanguageId();

        // 游标分页查询 10 条数据
        List<Long> quizCursorPageList = quizDOMapper.findQuizCursorPageList(cursor, languageId);

        // 从数据库中没有查询到问题，列表数据为空，直接返回
        if (CollUtil.isEmpty(quizCursorPageList)) {
            return Response.success();
        }

        // 构建 Redis 缓存的 key
        List<String> quizRedisKeys = quizCursorPageList.stream()
                .map(RedisKeyConstants::buildQuizBriefKey).toList();

        // 先从 Redis 缓存中查, multiGet 批量查询提升性能
        List<String> quizRedisValues = redisTemplate.opsForValue().multiGet(quizRedisKeys);

        // 如果缓存中不为空
        if (CollUtil.isNotEmpty(quizRedisValues)) {
            // 过滤掉为空的数据
            quizRedisValues = quizRedisValues.stream().filter(Objects::nonNull).toList();
        }

        // 反参
        List<QuizItemRspVO> quizItemRspVOS = Lists.newArrayList();

        // 将过滤后的缓存集合，转换为 VO 返参实体类
        if (CollUtil.isNotEmpty(quizRedisValues)) {
            quizItemRspVOS = quizRedisValues.stream()
                    .map(value -> JsonUtils.parseObject(value, QuizItemRspVO.class))
                    .collect(Collectors.toList());
        }

        // 如果被查询的笔记信息，都在 Redis 缓存中, 则直接返回
        if (CollUtil.size(quizCursorPageList) == CollUtil.size(quizItemRspVOS)) {
            // 找 ID 最小的值为游标
            Optional<Long> earliestId = quizItemRspVOS.stream()
                    .map(QuizItemRspVO::getId)
                    .min(Long::compareTo);

            // 构建返参
            FindQuizzesRspVO findQuizzesRspVO = FindQuizzesRspVO.builder()
                    .quizzes(quizItemRspVOS)
                    .nextCursor(earliestId.orElse(null))
                    .build();

            return Response.success(findQuizzesRspVO);
        }

        // 还有另外两种情况：一种是缓存里没有问题信息数据，还有一种是缓存里数据不全，需要从数据库中补充
        // 筛选出缓存里没有的问题数据，去查数据库
        List<Long> quizIdsNeedQuery;

        if (CollUtil.isNotEmpty(quizRedisValues)) {
            // 将 QuizItemRspVOS 集合转 Map
            Map<Long, QuizItemRspVO> map = quizItemRspVOS.stream()
                    .collect(Collectors.toMap(QuizItemRspVO::getId, p -> p));

            // 筛选出 Redis 缓存中不存在的 ID
            quizIdsNeedQuery = quizCursorPageList.stream()
                    .filter(id -> Objects.isNull(map.get(id)))
                    .toList();
        }
        // 缓存中一条用户信息都没查到，则提交的问题 ID 集合都需要查数据库
        else {
            quizIdsNeedQuery = quizCursorPageList;
        }

        // 根据 ID 批量查询问题，获取提问者的用户头像、昵称
        List<QuizDO> quizDOS = quizDOMapper.findQuizzesByIds(quizIdsNeedQuery);

        // 筛选出作者 ID，可能会存在相同作者的情况进行去重
        List<Long> quizCreatorIds = quizDOS.stream()
                .map(QuizDO::getCreatorId)
                .distinct()
                .toList();

        // 并发查询优化
        // RPC: 调用用户服务
        CompletableFuture<List<FindUserByIdRspDTO>> userResultFuture = CompletableFuture.supplyAsync(() ->
                userRpcService.findByIds(quizCreatorIds), threadPoolTaskExecutor);

        // RPC: 调用计数服务
        CompletableFuture<List<FindQuizCountsByIdRspDTO>> countResultFuture = CompletableFuture.supplyAsync(() ->
                countRpcService.findByQuizIds(quizIdsNeedQuery), threadPoolTaskExecutor);

        CompletableFuture<List<QuizItemRspVO>> resultFuture = CompletableFuture
                .allOf(userResultFuture, countResultFuture)
                .thenApply(s -> {
                    List<FindUserByIdRspDTO> findUserByIdRspDTOS = userResultFuture.join();
                    List<FindQuizCountsByIdRspDTO> findQuizCountsByIdRspDTOS = countResultFuture.join();

                    // 初始化固定长度数组
                    List<QuizItemRspVO> quizItemRspVOS2 = Lists.newArrayListWithCapacity(10);

                    if (CollUtil.isNotEmpty(findUserByIdRspDTOS) && CollUtil.isNotEmpty(findQuizCountsByIdRspDTOS)) {
                        // 根据用户 ID 分组
                        Map<Long, FindUserByIdRspDTO> findUserByIdRspDTOMap = findUserByIdRspDTOS.stream()
                                .collect(Collectors.toMap(FindUserByIdRspDTO::getId, Function.identity()));

                        // DTO 集合转 Map
                        Map<Long, FindQuizCountsByIdRspDTO> noteIdAndDTOMap = findQuizCountsByIdRspDTOS.stream()
                                .collect(Collectors.toMap(FindQuizCountsByIdRspDTO::getQuizId, Function.identity()));

                        // 需要查询问题关系的问题ID
                        List<Long> needQueryQuizIds = findQuizCountsByIdRspDTOS.stream()
                                .filter(object -> Objects.isNull(object.getFansTotal()) || object.getFansTotal() > 0)
                                .map(FindQuizCountsByIdRspDTO::getQuizId)
                                .toList();

                        // 问题 ID 和 粉丝头像
                        Map<Long, List<String>> quizIdAndFansImageMap = Maps.newHashMap();

                        // 当问题存在粉丝时才调用关系服务
                        if (CollUtil.isNotEmpty(needQueryQuizIds)) {
                            // 调用 RPC : 关系服务
                            FindQuizFansTop3ByIdsRspDTO quizFansTop3ByIds = relationRpcService.findQuizFansTop3ByIds(needQueryQuizIds);

                            quizIdAndFansImageMap.putAll(quizFansTop3ByIds.getQuizIdAndFansImageMap());
                        }

                        for (QuizDO quizDO : quizDOS) {
                            // 获取作者信息
                            FindUserByIdRspDTO findUserByIdRspDTO = findUserByIdRspDTOMap.get(quizDO.getCreatorId());
                            // 获取问题计数信息
                            FindQuizCountsByIdRspDTO findQuizCountsByIdRspDTO = noteIdAndDTOMap.get(quizDO.getId());
                            // 获取问题关系信息
                            List<String> fansImage = quizIdAndFansImageMap.get(quizDO.getId());

                            QuizItemRspVO findQuizzesRspVO = QuizItemRspVO.builder()
                                    .id(quizDO.getId())
                                    .title(quizDO.getTitle())
                                    .creatorId(quizDO.getCreatorId())
                                    .creatorName(findUserByIdRspDTO.getDisplayName())
                                    .avatar(findUserByIdRspDTO.getPhotoUrl())
                                    .nativeLanguage(findUserByIdRspDTO.getNativeLanguage())
                                    .interestLanguage(findUserByIdRspDTO.getInterestLanguage())
                                    .likeTotal(findQuizCountsByIdRspDTO.getLikeTotal())
                                    .fansTotal(findQuizCountsByIdRspDTO.getFansTotal())
                                    .browseTotal(findQuizCountsByIdRspDTO.getBrowseTotal())
                                    .commentTotal(findQuizCountsByIdRspDTO.getCommentTotal())
                                    .fansImages(fansImage)
                                    .build();

                            quizItemRspVOS2.add(findQuizzesRspVO);
                        }
                    }

                    return quizItemRspVOS2;
                });

        // 获取拼装后的 quizItemRspVOS2
        List<QuizItemRspVO> quizItemRspVOS2 = resultFuture.get();

        // 异步线程将用户信息同步到 Redis 中
        threadPoolTaskExecutor.submit(() -> {
            // 执行 pipeline 操作
            redisTemplate.executePipelined((RedisCallback<Object>) connection -> {
                for (QuizItemRspVO quizItemRspVO : quizItemRspVOS2) {
                    // 问题简要缓存 Redis Key
                    String quizBriefRedisKey = RedisKeyConstants.buildQuizBriefKey(quizItemRspVO.getId());

                    String value = JsonUtils.toJsonString(quizItemRspVO);
                    // 过期时间（保底1天 + 随机秒数，将缓存过期时间打散，防止同一时间大量缓存失效，导致数据库压力太大）
                    long expireSeconds = 60 * 60 * 24 + RandomUtil.randomInt(60 * 60 * 24);
                    // redisTemplate.opsForValue().set(quizBriefRedisKey, value, expireSeconds, TimeUnit.SECONDS);
                    redisTemplate.opsForValue().setIfAbsent(quizBriefRedisKey, value, expireSeconds, TimeUnit.SECONDS);
                }
                return null;
            });
        });

        // 合并数据
        if (CollUtil.isNotEmpty(quizItemRspVOS2)) {
            quizItemRspVOS.addAll(quizItemRspVOS2);
        }

        // 找 ID 最小的值为游标
        Optional<Long> earliestId = quizItemRspVOS.stream()
                .map(QuizItemRspVO::getId)
                .min(Long::compareTo);

        // 构建返参
        FindQuizzesRspVO findQuizzesRspVO = FindQuizzesRspVO.builder()
                .quizzes(quizItemRspVOS)
                .nextCursor(earliestId.orElse(null))
                .build();

        return Response.success(findQuizzesRspVO);
    }

    /**
     * 查询是否笔记存在
     *
     * @param checkNoteIsExistReqDTO
     * @return
     */
    @Override
    public Response<CheckQuizIsExistRespDTO> checkNoteIsExist(CheckNoteIsExistReqDTO checkNoteIsExistReqDTO) {
        // 获取问题 ID
        Long quizId = checkNoteIsExistReqDTO.getQuizId();

        // 先查询本地缓存
        String quizLocalCache = LOCAL_CACHE.getIfPresent(quizId);

        // 解析本地缓存
        FindQuizDetailRspVO findQuizDetailRspVO = JsonUtils.parseObject(quizLocalCache, FindQuizDetailRspVO.class);

        // 构建返参
        CheckQuizIsExistRespDTO checkQuizIsExistRespDTO = new CheckQuizIsExistRespDTO();

        // 本地缓存不存在
        if (Objects.isNull(findQuizDetailRspVO)) {
            // 构建 Redis Key
            String buildQuizDetailRedisKey = RedisKeyConstants.buildQuizDetailKey(quizId);

            // 查询 Redis
            String quizDetailRedisValue = redisTemplate.opsForValue().get(buildQuizDetailRedisKey);

            findQuizDetailRspVO = JsonUtils.parseObject(quizDetailRedisValue, FindQuizDetailRspVO.class);

            // redis 不存在
            if (Objects.isNull(findQuizDetailRspVO)) {

                // 查询数据库
                int count = quizDOMapper.selectCountByQuizId(quizId);

                // 如果数据库中不存在
                if (count == 0) {
                    throw new BizException(ResponseCodeEnum.QUIZ_NOT_FOUND);
                }

                // 数据库存在，异步同步到 redis 中
                threadPoolTaskExecutor.submit(() -> {
                    FindQuizDetailReqVO quizDetailReqVO = FindQuizDetailReqVO.builder()
                            .id(quizId)
                            .build();
                    findQuizDetail(quizDetailReqVO);
                });
            }
        }

        // 构建返参
        checkQuizIsExistRespDTO.setIsExist(Boolean.TRUE);
        return Response.success(checkQuizIsExistRespDTO);
    }

    /**
     * 点赞问题
     *
     * @param likeQuizReqVO
     * @return
     */
    @Override
    public Response<?> likeQuiz(LikeQuizReqVO likeQuizReqVO) {
        // 问题 ID
        Long quizId = likeQuizReqVO.getId();

        // 1. 校验问题是否存在, 若存在，则获取发布者用户 ID
        Long creatorId = checkNoteIsExistAndGetCreatorId(quizId);

        // 获取用户 ID
        Long userId = LoginUserContextHolder.getUserId();

        // 2. 判断目标笔记，是否已经点赞过
        String rbitmapUserQuizLikeListKey = RedisKeyConstants.buildRBitmapUserQuizLikeListKey(userId);

        DefaultRedisScript<Long> script = new DefaultRedisScript<>();
        // Lua 脚本路径
        script.setScriptSource(new ResourceScriptSource(new ClassPathResource("/lua/rbitmap_quiz_like_check.lua")));
        // 返回值类型
        script.setResultType(Long.class);

        // 执行 Lua 脚本，拿到返回结果
        Long result = redisTemplate.execute(script, Collections.singletonList(rbitmapUserQuizLikeListKey), quizId);

        log.info("Lua 返回结果：{}", result);

        QuizLikeLuaResultEnum quizLikeLuaResultEnum = QuizLikeLuaResultEnum.valueOf(result);

        if (Objects.isNull(quizLikeLuaResultEnum)) {
            throw new RuntimeException("Lua 返回结果错误");
        }

        switch (quizLikeLuaResultEnum) {
            // Redis 中 rbitmap 不存在
            case R_BITMAP_NOT_EXIST -> {
                int count = quizLikeDOMapper.selectCountByUserIdAndQuizId(userId, quizId);

                // 保底1天 + 随机秒数
                long expireSeconds = 60 * 60 * 24 + RandomUtil.randomInt(60 * 60 * 24);

                // 目标问题已被点赞
                if (count > 0) {
                    threadPoolTaskExecutor.submit(() -> batchAddQuizLike2RBitmapAndExpire(userId, expireSeconds, rbitmapUserQuizLikeListKey));
                    throw new BizException(ResponseCodeEnum.QUIZ_ALREADY_LIKED);
                }

                // 若目标问题未被点赞，查询当前用户是否有点赞其他问题，有则同步初始化 Roaring Bitmap
                batchAddQuizLike2RBitmapAndExpire(userId, expireSeconds, rbitmapUserQuizLikeListKey);

                // Lua 脚本路径
                script.setScriptSource(new ResourceScriptSource(new ClassPathResource("/lua/rbitmap_add_quiz_like_and_expire.lua")));
                // 返回值类型
                script.setResultType(Long.class);
                // 执行 rbitmap_quiz_like_and_expire.lua 脚本，拿到返回结果
                redisTemplate.execute(script, Collections.singletonList(rbitmapUserQuizLikeListKey), quizId, expireSeconds);
            }
            // 目标笔记已经被点赞
            case QUIZ_LIKED -> throw new BizException(ResponseCodeEnum.QUIZ_ALREADY_LIKED);
        }

        // 3. 发送 MQ， 将点赞数据落库
        // 构建消息体 DTO
        LikeUnlikeQuizMqDTO likeUnlikeQuizMqDTO = LikeUnlikeQuizMqDTO.builder()
                .userId(userId)
                .quizId(quizId)
                // 点赞问题
                .type(LikeUnlikeQuizTypeEnum.LIKE.getCode())
                .createTime(LocalDateTime.now())
                // 问题发布者 ID
                .quizCreatorId(creatorId)
                .build();

        // 构建消息对象，并将 DTO 转成 Json 字符串设置到消息体中
        Message<String> message = MessageBuilder.withPayload(JsonUtils.toJsonString(likeUnlikeQuizMqDTO))
                .build();

        // 通过冒号连接, 可让 MQ 发送给主题 Topic 时，携带上标签 Tag
        String destination = MQConstants.TOPIC_LIKE_OR_UNLIKE + ":" + MQConstants.TAG_LIKE;

        String hashKey = String.valueOf(userId);

        // 异步发送 MQ 消息，提升接口响应速度
        rocketMQTemplate.asyncSendOrderly(destination, message, hashKey, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("==> 【问题点赞】MQ 发送成功，SendResult: {}", sendResult);
            }

            @Override
            public void onException(Throwable throwable) {
                log.error("==> 【问题点赞】MQ 发送异常: ", throwable);
            }
        });

        return Response.success();
    }

    /**
     * 取消点赞问题
     *
     * @param unlikeQuizReqVO
     * @return
     */
    @Override
    public Response<?> unlikeQuiz(UnlikeQuizReqVO unlikeQuizReqVO) {
        // 问题ID
        Long quizId = unlikeQuizReqVO.getId();

        // 1. 校验问题是否存在
        Long creatorId = checkNoteIsExistAndGetCreatorId(quizId);

        // 2.校验用户是否已经点赞过

        // 用户 ID
        Long userId = LoginUserContextHolder.getUserId();

        String rBitmapUserQuizLikeListKey = RedisKeyConstants.buildRBitmapUserQuizLikeListKey(userId);

        DefaultRedisScript<Long> script = new DefaultRedisScript<>();
        // Lua 脚本路径
        script.setScriptSource(new ResourceScriptSource(new ClassPathResource("/lua/rbitmap_quiz_unlike_check.lua")));
        // 返回值类型
        script.setResultType(Long.class);
        Long result = redisTemplate.execute(script, Collections.singletonList(rBitmapUserQuizLikeListKey), quizId);

        QuizLikeLuaResultEnum quizLikeLuaResultEnum = QuizLikeLuaResultEnum.valueOf(result);

        if (Objects.isNull(quizLikeLuaResultEnum)) {
            throw new RuntimeException("Lua 脚本返回结果错误");
        }

        switch (quizLikeLuaResultEnum) {
            // Roaring Bitmap 不存在
            case R_BITMAP_NOT_EXIST -> {
                // 从数据库中校验笔记是否被点赞
                int count = quizLikeDOMapper.selectCountByUserIdAndQuizId(userId, quizId);

                // 保底1天 + 随机秒数
                long expireSeconds = 60 * 60 * 24 + RandomUtil.randomInt(60 * 60 * 24);

                // 未点赞，无法取消点赞操作，抛出业务异常
                if (count == 0) {
                    threadPoolTaskExecutor.submit(() -> batchAddQuizLike2RBitmapAndExpire(userId, expireSeconds, rBitmapUserQuizLikeListKey));
                    throw new BizException(ResponseCodeEnum.QUIZ_NOT_LIKED);
                }

                // 初始化 Roaring Bitmap
                batchAddQuizLike2RBitmapAndExpire(userId, expireSeconds, rBitmapUserQuizLikeListKey);
                // Lua 脚本路径
                script.setScriptSource(new ResourceScriptSource(new ClassPathResource("/lua/rbitmap_add_quiz_unlike_and_expire.lua")));
                // 返回值类型
                script.setResultType(Long.class);
                // 执行 rbitmap_add_quiz_unlike_and_expire 脚本 取消问题点赞
                redisTemplate.execute(script, Collections.singletonList(rBitmapUserQuizLikeListKey), quizId, expireSeconds);
            }
            // 目标笔记未被点赞
            case QUIZ_NOT_LIKED -> throw new BizException(ResponseCodeEnum.QUIZ_NOT_LIKED);
        }

        // 3.发送 MQ, 数据更新落库
        // 构建消息体 DTO
        LikeUnlikeQuizMqDTO likeUnlikeQuizMqDTO = LikeUnlikeQuizMqDTO.builder()
                .userId(userId)
                .quizId(quizId)
                // 取消点赞问题
                .type(LikeUnlikeQuizTypeEnum.UNLIKE.getCode())
                .createTime(LocalDateTime.now())
                // 问题发布者 ID
                .quizCreatorId(creatorId)
                .build();

        // 构建消息对象，并将 DTO 转成 Json 字符串设置到消息体中
        Message<String> message = MessageBuilder.withPayload(JsonUtils.toJsonString(likeUnlikeQuizMqDTO)).build();

        // 通过冒号连接, 可让 MQ 发送给主题 Topic 时，携带上标签 Tag
        String destination = MQConstants.TOPIC_LIKE_OR_UNLIKE + ":" + MQConstants.TAG_UNLIKE;

        String hashKey = String.valueOf(userId);

        // 异步发送 MQ 消息，提升接口响应速度
        rocketMQTemplate.asyncSendOrderly(destination, message, hashKey, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("==> 【问题取消点赞】MQ 发送成功，SendResult: {}", sendResult);
            }

            @Override
            public void onException(Throwable throwable) {
                log.error("==> 【笔记取消点赞】MQ 发送异常: ", throwable);
            }
        });

        return Response.success();
    }

    /**
     * 初始化问题点赞 Roaring Bitmap
     *
     * @param userId
     * @param expireSeconds
     * @param rbitmapUserQuizLikeListKey
     */
    private void batchAddQuizLike2RBitmapAndExpire(Long userId, Long expireSeconds, String rbitmapUserQuizLikeListKey) {
        try {
            // 异步全量同步一下，并设置过期时间
            List<QuizLikeDO> quizLikeDOS = quizLikeDOMapper.selectByUserId(userId);

            if (CollUtil.isNotEmpty(quizLikeDOS)) {
                DefaultRedisScript<Long> script = new DefaultRedisScript<>();
                // Lua 脚本路径
                script.setScriptSource(new ResourceScriptSource(new ClassPathResource("/lua/rbitmap_batch_add_quiz_like_and_expire.lua")));
                // 返回值类型
                script.setResultType(Long.class);

                // 构建 Lua 参数
                List<Long> luaArgs = Lists.newArrayList();
                quizLikeDOS.forEach(quizLikeDO -> luaArgs.add(quizLikeDO.getQuizId()));
                // 最后一个参数是过期时间（秒）
                luaArgs.add(expireSeconds);
                redisTemplate.execute(script, Collections.singletonList(rbitmapUserQuizLikeListKey), luaArgs.toArray());
            }
        } catch (Exception e) {
            log.error("## 异步初始化【问题点赞】Roaring Bitmap 异常: ", e);
        }
    }

    /**
     * 校验问题是否存在，若存在，则获取问题的发布者 ID
     *
     * @param quizId
     * @return
     */
    private Long checkNoteIsExistAndGetCreatorId(Long quizId) {
        // 先查询本地缓存
        String quizLocalCache = LOCAL_CACHE.getIfPresent(quizId);

        // 解析本地缓存
        FindQuizDetailRspVO findQuizDetailRspVO = JsonUtils.parseObject(quizLocalCache, FindQuizDetailRspVO.class);

        // 本地缓存不存在
        if (Objects.isNull(findQuizDetailRspVO)) {
            // 构建 Redis Key
            String buildQuizDetailRedisKey = RedisKeyConstants.buildQuizDetailKey(quizId);

            // 查询 Redis
            String quizDetailRedisValue = redisTemplate.opsForValue().get(buildQuizDetailRedisKey);

            findQuizDetailRspVO = JsonUtils.parseObject(quizDetailRedisValue, FindQuizDetailRspVO.class);

            // redis 不存在
            if (Objects.isNull(findQuizDetailRspVO)) {
                // 问题发布者用户 ID
                Long creatorId = quizDOMapper.selectCreatorIdByQuizId(quizId);

                // 如果数据库中不存在
                if (Objects.isNull(creatorId)) {
                    throw new BizException(ResponseCodeEnum.QUIZ_NOT_FOUND);
                }

                // 数据库存在，异步同步到 redis 中
                threadPoolTaskExecutor.submit(() -> {
                    FindQuizDetailReqVO quizDetailReqVO = FindQuizDetailReqVO.builder()
                            .id(quizId)
                            .build();
                    findQuizDetail(quizDetailReqVO);
                });

                return creatorId;
            }
        }

        return findQuizDetailRspVO.getCreatorId();
    }

    /**
     * 获取是否点赞数据
     *
     * @param findQuizIsLikeReqVO
     * @return
     */
    @Override
    public Response<FindQuizIsLikeRspVO> isLikedData(FindQuizIsLikeReqVO findQuizIsLikeReqVO) {
        Long quizId = findQuizIsLikeReqVO.getQuizId();

        // 已登录的用户 ID
        Long currUserId = LoginUserContextHolder.getUserId();

        // 默认未点赞
        boolean isLiked = false;

        // 若当前用户已经登录
        if (Objects.nonNull(currUserId)) {
            // 校验是否点赞
            isLiked = checkQuizIsLiked(quizId, currUserId);
        }

        return Response.success(FindQuizIsLikeRspVO.builder()
                .quizId(quizId)
                .isLiked(isLiked)
                .build());
    }

    /**
     * 查询问题作者信息
     *
     * @param findQuizCreatorReqDTO
     * @return
     */
    @Override
    public Response<FindQuizCreatorRespDTO> findQuizCreator(FindQuizCreatorReqDTO findQuizCreatorReqDTO) {
        // 获取问题 ID
        Long quizId = findQuizCreatorReqDTO.getQuizId();

        // 校验问题是否存在并获取作者 ID
        Long creatorId = checkNoteIsExistAndGetCreatorId(quizId);

        // 构建返回参数
        FindQuizCreatorRespDTO findQuizCreatorRespDTO = FindQuizCreatorRespDTO.builder()
                .creatorId(creatorId)
                .build();

        return Response.success(findQuizCreatorRespDTO);
    }

    /**
     * 校验当前用户是否点赞笔记
     *
     * @param quizId
     * @param currUserId
     * @return
     */
    private boolean checkQuizIsLiked(Long quizId, Long currUserId) {
        // 是否点赞
        boolean isLiked = false;

        // Roaring Bitmap Key
        String rbitmapUserNoteLikeListKey = RedisKeyConstants.buildRBitmapUserQuizLikeListKey(currUserId);

        DefaultRedisScript<Long> script = new DefaultRedisScript<>();
        // Lua 脚本路径
        script.setScriptSource(new ResourceScriptSource(new ClassPathResource("/lua/rbitmap_quiz_like_only_check.lua")));
        // 返回值类型
        script.setResultType(Long.class);

        // 执行 Lua 脚本，拿到返回结果
        Long result = redisTemplate.execute(script, Collections.singletonList(rbitmapUserNoteLikeListKey), quizId);

        QuizLikeLuaResultEnum quizLikeLuaResultEnum = QuizLikeLuaResultEnum.valueOf(result);

        switch (quizLikeLuaResultEnum) {
            // Redis 中 Roaring Bitmap 不存在
            case R_BITMAP_NOT_EXIST -> {
                // 从数据库中校验问题是否被点赞，并异步初始化 Roaring Bitmap，设置过期时间
                int count = quizLikeDOMapper.selectCountByUserIdAndQuizId(currUserId, quizId);

                // 保底 1 天 + 随机秒数
                long expireSeconds = 60 * 60 * 24 + RandomUtil.randomInt(60 * 60 * 24);

                // 目标问题已被点赞
                if (count > 0) {
                    // 异步初始化 Roaring Bitmap
                    threadPoolTaskExecutor.submit(() ->
                            batchAddQuizLike2RBitmapAndExpire(currUserId, expireSeconds, rbitmapUserNoteLikeListKey));
                    isLiked = true;
                }
            }
            // Roaring Bitmap 判断已点赞
            case QUIZ_LIKED -> isLiked = true;
        }

        return isLiked;
    }
}
